package com.atguigu.day04;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
//import org.apache.phoenix.queryserver.client.ThinClientUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class Flink02_Sink_Custom {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.从端口读取数据，并转为JavaBean
        SingleOutputStreamOperator<WaterSensor> dataStream = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                    }
                });

        //TODO 3.自定义Sink将数据写入Phoenix
        dataStream.addSink(new MySink());
        env.execute();
    }

    //使用富函数优化链接个数
//    public static class MySink implements SinkFunction<WaterSensor>{
    public static class MySink extends RichSinkFunction<WaterSensor> {

        private Connection connection = null;
        private PreparedStatement pstm = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("创建链接");
            //获取连接
//            String url = ThinClientUtil.getConnectionUrl("hadoop102", 8765);
            connection = DriverManager.getConnection("111");

        }

        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {
            //获取语句预执行
            pstm = connection.prepareStatement("upsert into STUDENT values(?,?,?)");
            pstm.setString(1, value.getId());
            pstm.setString(2, value.getId() + "zhangsan");
            pstm.setString(3, value.getId() + "bj");

            pstm.execute();
            connection.commit();

        }

        @Override
        public void close() throws Exception {
            System.out.println("关闭连接");
            pstm.close();
            connection.close();
        }
    }
}
